
#include "./worker.h"

namespace lukv {


worker::worker(int _id) : id(_id), sock(socket(AF_INET, SOCK_STREAM, 0)), paddr(&manager::worker_addrs[_id]) { }

#ifdef __OSX__
int worker::start() {
    const int BACKLOG_SIZE = 16;
    const int READ_BUF_SIZE = 2048;
    const int K_EVENTS_CNT = 128;

    set_fd_noblk(sock);

    if (bind(sock, (sockaddr *)paddr, sizeof(*paddr)) < 0) {
        close_fds(sock);
        EXIT_ERR(strerror(errno));
    }

    if (listen(sock, BACKLOG_SIZE)) {
        close_fds(sock);
        EXIT_ERR(strerror(errno));
    }

    int kq = kqueue();

    struct kevent read_event { (uint64_t) sock, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, nullptr };

    if (kevent(kq, &read_event, 1, nullptr, 0, nullptr) < 0) {
        close_fds(sock, kq);
        EXIT_ERR(strerror(errno));
    }

    struct kevent out_events[K_EVENTS_CNT];

    char read_buf[READ_BUF_SIZE];

    timespec over_t { 0, 500000000 };

    while (true) {
        int cnt = kevent(kq, nullptr, 0, out_events, K_EVENTS_CNT, &over_t);
        if (cnt < 0) {
            LOG_ERR(strerror(errno));
            break;
        }
        else {
            for (int i = 0; i < cnt; ++i) {
                auto _sock = out_events[i].ident;
                auto _data = out_events[i].data;

                if (out_events[i].filter & EVFILT_READ) {
                    println("[READ_EV]", "[ident:", _sock, ']', "[data:", _data, ']');

                    if (_sock == sock) {
                        sockaddr_in cli_addr { };
                        socklen_t  cli_addr_len = sizeof(cli_addr);

                        for (int j = 0; j < _data; ++j) {
                            int cli_sock = accept(sock, (sockaddr *) &cli_addr, &cli_addr_len);
                            if (cli_sock < 0) {
                                LOG_ERR(strerror(errno));
                                continue;
                            }

                            set_fd_noblk(cli_sock);

                            println("[sock:", cli_sock, ']', "[client:", inet_ntoa(cli_addr.sin_addr), ':', ntohs(cli_addr.sin_port), ']');

                            if (!manager::is_node(cli_addr)) {
                                close(cli_sock);
                                continue;
                            }

                            struct kevent cli_rw_events { (uintptr_t) cli_sock, EVFILT_READ | EVFILT_WRITE, EV_ENABLE | EV_ADD, 0, 0, nullptr };
                            kevent(kq, &cli_rw_events, 1, nullptr, 0, nullptr);
                        }
                    }
                    else {
                        auto rlen = recv(_sock, read_buf, _data, 0);

                        if (rlen > 0) {
                            std::string read_str(read_buf, rlen);
                            println("[recv]<", read_str, '>');

                            std::string wt_msg = handle_cmd(read_str);

                            send(_sock, wt_msg.c_str(), wt_msg.size(), 0);
                            println("[send]<", wt_msg, '>');

                            if (read_str == "#exit") {
                                close_fds(_sock);
                                struct kevent cli_event { (uintptr_t) _sock, EVFILT_VNODE, EV_DISABLE | EV_DELETE, 0, 0, nullptr };
                                kevent(kq, &cli_event, 1, nullptr, 0, nullptr);
                                manager::remove_expire_cli(_sock);
                                println("closed fd: ", _sock);
                                continue;
                            }
                        }
                        else {
                            close_fds(_sock);
                            struct kevent cli_event { (uintptr_t) _sock, EVFILT_VNODE, EV_DISABLE | EV_DELETE, 0, 0, nullptr };
                            kevent(kq, &cli_event, 1, nullptr, 0, nullptr);
                            manager::remove_expire_cli(_sock);
                            println("closed fd: ", _sock);
                            continue;
                        }

                        struct kevent cli_rw_events { (uintptr_t) _sock, EVFILT_READ | EVFILT_WRITE, EV_ENABLE | EV_ADD, 0, 0, nullptr };
                        kevent(kq, &cli_rw_events, 1, nullptr, 0, nullptr);
                    }
                }
            }
        }
    }

    return 0;
}
#endif

#ifdef __LINUX__
int worker::start() {

    const int EP_EVENT_CNT = 16;
    const int BACK_LOG_SIZE = 32;
    const int READ_BUF_SIZE = 1024;

    set_fd_noblk(sock);

    if (bind(sock, (sockaddr *)paddr, sizeof(*paddr)) < 0) {
        close_fds(sock);
        EXIT_ERR(strerror(errno));
    }

    if (listen(sock, BACK_LOG_SIZE) < 0) {
        close_fds(sock);
        EXIT_ERR(strerror(errno));
    }

    int ep_fd = epoll_create(64);

    struct epoll_event srv_event { };
    srv_event.events = EPOLLIN;
    srv_event.data.fd = sock;

    epoll_ctl(ep_fd, EPOLL_CTL_ADD, sock, &srv_event);

    struct epoll_event ep_events[EP_EVENT_CNT];

    char read_buf[READ_BUF_SIZE];

    while (true) {
        int cnt = epoll_wait(ep_fd, ep_events, EP_EVENT_CNT, 500);
        if (cnt < 0) {
            LOG_ERR(strerror(errno));
            break;
        }
        
        for (int i = 0; i < cnt; i++) {
            int _sock = ep_events[i].data.fd;

            if (_sock == sock) {
                struct sockaddr_in cli_addr { };
                socklen_t cli_addr_len = sizeof(cli_addr);

                int cli_sock = accept(sock, (sockaddr *)&cli_addr, &cli_addr_len);
                if (cli_sock < 0) {
                    LOG_ERR(strerror(errno));
                    continue;
                }

                set_fd_noblk(cli_sock);

                println("[sock:", cli_sock, ']', "[client:", inet_ntoa(cli_addr.sin_addr), ':', ntohs(cli_addr.sin_port), ']');

                srv_event.events = EPOLLIN | EPOLLET;
                srv_event.data.fd = cli_sock;

                epoll_ctl(ep_fd, EPOLL_CTL_ADD, cli_sock, &srv_event);
            }
            else {
                char * _ptr_r = read_buf;

                int r_len = recv(_sock, read_buf, READ_BUF_SIZE, 0);

                if (r_len <= 0) {
                    close_fds(_sock);
                    epoll_ctl(ep_fd, EPOLL_CTL_DEL, _sock, nullptr);
                    manager::remove_expire_cli(_sock);
                    println("closed fd: ", _sock);
                    continue;
                }
                else {
                    _ptr_r += r_len;
                    while (r_len = recv(_sock, _ptr_r, READ_BUF_SIZE, 0) > 0) {
                        _ptr_r += r_len;
                    }
                }

                std::string r_msg(read_buf, _ptr_r);
                println("[recv]<", r_msg, '>');

                std::string w_msg = handle_cmd(r_msg);
                send(_sock, w_msg.c_str(), w_msg.size(), 0);
                println("[send]<", w_msg, '>');

                if (r_msg == "#exit") {
                    close_fds(_sock);
                    epoll_ctl(ep_fd, EPOLL_CTL_DEL, _sock, nullptr);
                    manager::remove_expire_cli(_sock);
                    println("closed fd: ", _sock);
                    continue;
                }
            }
        }
    }

    return 0;
}
#endif


std::string worker::handle_cmd(const std::string & _cmd) {
    // worker这边相较于driver更简单，因为worker只负责执行具体的key操作，并将操作结果返回到driver
    // worker不接受客户端发起的连接，若接收到来自客户端的连接，则直接关闭socket
    std::pair<char, std::string> cur = base.handle_f(_cmd.substr(1));
    std::string ans;
    if (cur.first < 0) {
        ans.push_back('E');
    }
    else {
        ans.push_back('S');
    }
    ans.push_back('|');
    ans.append(cur.second);
    return ans;
}

}
